package bx.actor;

import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.http.javadsl.Http;
import akka.http.javadsl.marshallers.jackson.Jackson;
import akka.http.javadsl.model.*;
import bx.InteractionEntity.CommonResponse;
import bx.InteractionEntity.WorkerHeartbeatRequest;
import bx.JobWorkerBootstrap;
import com.google.gson.Gson;
import lombok.Builder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;


public class WorkerHeartbeatActor extends AbstractBehavior<WorkerHeartbeatActor.WorkerHeartbeatActorMsg> {
    private final static Logger log = LoggerFactory.getLogger(WorkerHeartbeatActor.class);
    public static final String HEARTBEAT_PATH = "/heartbeat";
    private int HeartbeatIntervalTime = 5; // 默认 5s 一次心跳
    private final List<String> bxJobServerList;
    private int currentIndex;

    private int failCount;

    interface WorkerHeartbeatActorMsg {
    }


    @Builder
    public static class WorkerHeartbeatStartMsg implements WorkerHeartbeatActorMsg {
    }

    public WorkerHeartbeatActor(ActorContext<WorkerHeartbeatActor.WorkerHeartbeatActorMsg> context, List<String> bxJobServerList) {
        super(context);
        this.bxJobServerList = bxJobServerList;
        int heartbeatTime = context.getSystem().settings().config().getInt("bx-job-worker.heartbeat-interval-time");
        if (heartbeatTime > 0) {
            this.HeartbeatIntervalTime = heartbeatTime;
        }
    }

    public static Behavior<WorkerHeartbeatActor.WorkerHeartbeatActorMsg> create(List<String> bxJobServerList) {
        return Behaviors.setup(context -> new WorkerHeartbeatActor(context, bxJobServerList));
    }

    @Override
    public Receive<WorkerHeartbeatActor.WorkerHeartbeatActorMsg> createReceive() {
        return newReceiveBuilder().onMessage(WorkerHeartbeatStartMsg.class, this::onStartHeartbeatMsg).build();
    }


    private Behavior<WorkerHeartbeatActorMsg> onStartHeartbeatMsg(WorkerHeartbeatStartMsg workerHeartbeatStartMsg) {
        if (failCount == this.bxJobServerList.size()) {
            failCount = 0;
            log.error("heartbeat failed count limited, server list:{}", bxJobServerList);
            JobWorkerBootstrap.getGlobalScheduledExecutorService().schedule(
                    () -> this.getContext().getSelf().tell(WorkerHeartbeatStartMsg.builder().build()), HeartbeatIntervalTime, TimeUnit.SECONDS);
            return this;
        }
        if (currentIndex == this.bxJobServerList.size()) {
            currentIndex = 0;
        }
        Gson gson = new Gson();
        String currentServer = this.bxJobServerList.get(currentIndex++); // 轮询执行心跳
        WorkerHeartbeatRequest workerHeartbeatRequest = new WorkerHeartbeatRequest();
        //todo 组装 WorkerHeartbeatRequest 信息
        HttpRequest httpRequest = HttpRequest.create(currentServer + HEARTBEAT_PATH).withMethod(HttpMethods.POST)
                .withEntity(HttpEntities.create(ContentTypes.APPLICATION_JSON, gson.toJson(workerHeartbeatRequest)));
        final CompletionStage<HttpResponse> responseFuture = Http.get(this.getContext().getSystem()).singleRequest(httpRequest);
        responseFuture.whenComplete((httpResponse, error) -> {
            if (error != null) {
                log.error("heartbeat failed, server:{}", currentServer, error);
                failCount++;
                this.getContext().getSelf().tell(WorkerHeartbeatStartMsg.builder().build());
                return;
            }
            CompletionStage<CommonResponse> unmarshal = Jackson.unmarshaller(CommonResponse.class).unmarshal(httpResponse.entity(), this.getContext().getSystem());
            unmarshal.toCompletableFuture().whenComplete((res, throwable) -> {
                if (throwable == null && res.getStatus() == CommonResponse.SUCCESS_STATUS) {
                    log.info("heartbeat success, server:{}", currentServer);
                    JobWorkerBootstrap.getGlobalScheduledExecutorService().schedule(
                            () -> this.getContext().getSelf().tell(WorkerHeartbeatStartMsg.builder().build()), HeartbeatIntervalTime, TimeUnit.SECONDS);
                } else {
                    failCount++;
                    log.info("heartbeat failed, server:{},response:{}", currentServer, res, throwable);
                    this.getContext().getSelf().tell(WorkerHeartbeatStartMsg.builder().build());
                }

            });
        });
        return this;
    }


}